
function isObject(variable) {
    return Object.prototype.toString.call(variable) === "[object Object]";
}

class Subject {
    subscribes = []
    subscribe = (obj) => {
        this.subscribes.push(obj)
    }
    next = (value) => {
        this.subscribes.forEach(item => {
            item.next(value)
        })
    }
}

const Rx = {
    Observable: {
        create: (subscribe_fn) => {
            return {
                scheduler: null,
                subscribe: (observer) => {
                    let subscribe_status = 1
                    const defaultOptions = {
                        error: err => console.error('Observer got an error: ' + err),
                        complete: () => console.log('Observer got a complete notification'),
                    }
                    const unsubscribe = () => {
                        subscribe_status = 0
                    }
                    let newObserver = {}
                    if (typeof observer === 'function') {
                        newObserver = {
                            next: observer
                        }
                    } else if (isObject(observer)) {
                        newObserver = observer
                    }
                    const mergeObserver = {
                        ...defaultOptions,
                        ...observer
                    }
                    subscribe_fn({
                        next: (val) => {
                            if (subscribe_status === 1) {
                                if (this.scheduler) {
                                    this.scheduler.schedule(
                                        (x) => mergeObserver.next(x),
                                        0 /* 延迟时间 */,
                                        val /* 会作为上面函数所使用的 x */
                                      );
                                } else {
                                    mergeObserver.next(val)
                                }
                            }
                        },
                        error: (err) => {
                            unsubscribe()
                            mergeObserver.error(err)
                        },
                        complete: () => {
                            unsubscribe()
                            mergeObserver.complete()
                        }
                    })
                    return {
                        unsubscribe: unsubscribe
                    }
                },
                observeOn(scheduler) {
                    this.scheduler = scheduler
                }
            }
        },
        interval(timer){
            return {
                subscribe: (observer) => {
                    let childrenSubscriptions = new Set()
                    let newTimer = 0
                    const interval = setInterval(() => {
                        newTimer++
                        observer(newTimer)
                    }, timer)
                    const unsubscribe = () => {
                        clearInterval(interval)
                        childrenSubscriptions.forEach(item => {
                            item.unsubscribe()
                        })
                    }
                    const add = (subscription) => {
                        childrenSubscriptions.add(subscription)
                    }
                    const remove = (subscription) => {
                        childrenSubscriptions.delete(subscription)
                    }
                    return {
                        unsubscribe,
                        add,
                        remove
                    }
                }
            }
        },
        from(array, scheduler){
            return {
                subscribe: (observer) => {
                    array.forEach(item => {
                        if (scheduler) {
                            scheduler.schedule(
                                (x) => observer.next(x),
                                0 /* 延迟时间 */,
                                val /* 会作为上面函数所使用的 x */
                              );
                        } else {
                            observer.next(item)
                        }
                    })
                },
                multicast: (observer) => {
                    return {
                        subscribe() {

                        }
                    }
                }
            }
        }
    },
    Subject,
    Scheduler: {
        async: {
            schedule(observerFn, timer, val) {
                setTimeout(() => {
                    observerFn(val)
                }, timer)
            }
        },
        asap: {},
        queue: {}
    }
}